package day

import (
	oceanApi "cid-media-api/api/ocean"
	"cid-media-api/global"
	"cid-media-api/models"
	. "github.com/oceanengine/ad_open_sdk_go/models"
	"github.com/robfig/cron/v3"
	"github.com/samber/lo"
	"gorm.io/gorm/clause"
	"log"
	"sync"
	"sync/atomic"
	"time"
)

func Run() {
	c := cron.New()
	c.AddFunc("*/60 * * * *", func() {
		startTime := time.Now().Format(time.DateTime)
		log.Printf("====================================================")
		log.Printf("正在定时执行广告主账号拉取任务")
		log.Printf("任务开始时间： %v", startTime)

		Execute(0, time.Now().Format(time.DateOnly), time.Now().Format(time.DateOnly))

		endTime := time.Now().Format(time.DateTime)
		log.Printf("任务结束时间： %v", endTime)
		log.Printf("====================================================")
	})

	c.Run()
}

func Execute(advertiserID int, startDate string, endDate string) {
	// 声明存放媒体账户的切片
	var mediaAccounts []models.MediaAccount

	insertCount := 0
	updateCount := 0

	// 查询数据库中的媒体账户，条件是状态为1且媒体ID符合要求
	db := global.CidDb.Where("status = ?", 1).Where("media_id = ?", oceanApi.MediaIdOcean)

	// 如果提供了advertiserID，进一步过滤查询结果
	if advertiserID > 0 {
		db = db.Where("advertiser_id = ?", advertiserID)
	}

	// 执行查询，并将结果存入mediaAccounts中，如果出错则打印错误并返回
	err := db.Find(&mediaAccounts).Error
	if err != nil {
		print(err)
		return
	}

	// 遍历每个媒体账户
	for _, account := range mediaAccounts {
		// 存放API返回的日销售报告数据
		var apiData []models.ReportDayAdvertiser
		apiDataChan := make(chan models.ReportDayAdvertiser) // 创建通道用于传递数据

		// 声明一个WaitGroup，用于等待所有goroutine完成
		var wg sync.WaitGroup

		const pSize int64 = 100 // 每页请求的大小

		var flag atomic.Bool     // 声明一个原子布尔值，用于控制循环
		flag.Store(true)         // 初始化为true，开始循环
		var flagMutex sync.Mutex // 声明一个互斥锁，用于控制flag的修改
		var page atomic.Int64    // 声明一个原子整型，用于表示当前请求的页数
		page.Store(1)            // 初始化为1，表示第一页

		for {
			// 如果flag被设置为false，退出循环
			flagMutex.Lock()
			if !flag.Load() {
				flagMutex.Unlock()
				break
			}

			// 启动一个新的goroutine来处理API请求
			go func() {
				pg := page.Load() // 加载当前页数
				// 调用API获取广告主报告数据
				res := oceanApi.GetAdvertiserReport(account, startDate, endDate, pg, pSize, STAT_TIME_GRANULARITY_DAILY_ReportAdvertiserGetV2TimeGranularity)

				// 如果返回结果为空或列表长度为0，停止循环
				if res == nil || len(res.List) == 0 {
					flag.Store(false) // 设置flag为false，退出循环
					flagMutex.Unlock()
					return
				}

				// 遍历返回的报告数据
				for _, item := range res.List {
					wg.Add(1)                                  // 增加WaitGroup计数
					apiDataChan <- models.ReportDayAdvertiser{ // 将数据传递到通道
						MediaId:        account.MediaId,
						CompanyId:      account.CompanyId,
						MediaAccountId: account.MediaAccountId,
						AdvertiserId:   account.AdvertiserId,
						AdvertiserName: account.AdvertiserName,
						Tdate:          *item.StatDatetime,
						Click:          *item.Click,
						PayAmountRoi:   *item.PayAmountRoi,
						ValidPlay:      *item.ValidPlay,
						StatPayAmount:  *item.StatPayAmount,
						Cost:           *item.Cost,
						ConvertCost:    *item.ConvertCost,
						Cpc:            *item.AvgClickCost,
						Show:           *item.Show,
						Ctr:            *item.Ctr,
						Convert:        *item.Convert,
						TotalPlay:      *item.TotalPlay,
						Cpm:            *item.AvgShowCost,
					}
				}

				totalPages := *res.PageInfo.TotalPage // 获取总页数
				if pg >= totalPages {                 // 如果当前页已达到或超过总页数，停止获取更多数据
					flag.Store(false) // 设置flag为false，退出循环
					flagMutex.Unlock()
					wg.Done()
					return
				}
				page.Add(1) // 增加页数，继续请求下一页
			}()
		}

		// 启动另一个goroutine用于收集从apiDataChan通道来的数据
		go func() {
			for data := range apiDataChan { // 从通道中接收数据
				apiData = append(apiData, data) // 将接收到的数据添加到apiData切片中

				wg.Done() // 完成一次数据处理，减少WaitGroup计数
			}
		}()

		wg.Wait()          // 等待所有相关的goroutine完成
		close(apiDataChan) // 确保在所有goroutine完成后关闭通道

		if len(apiData) == 0 { // 如果apiData切片为空，则跳过本次循环
			continue
		}

		var existedRecords []models.ReportDayAdvertiser
		advertiserIds := lo.Map(apiData, func(item models.ReportDayAdvertiser, idx int) int64 {
			return item.AdvertiserId
		})

		db.Where("advertiser_id IN (?)", advertiserIds).
			Where("tdate = ?", startDate).
			Where("media_id = ?", oceanApi.MediaIdOcean).
			Find(&existedRecords)

		var recordsToInsert []models.ReportDayAdvertiser
		var recordsToUpdate []models.ReportDayAdvertiser

		for _, item := range apiData {
			if lo.SomeBy(existedRecords, func(existed models.ReportDayAdvertiser) bool {
				return existed.AdvertiserId == item.AdvertiserId &&
					existed.Tdate == item.Tdate &&
					existed.MediaId == item.MediaId
			}) {
				// 如果存在相同的记录，则添加进更新列表
				recordsToInsert = append(recordsToInsert, item)
			} else {
				// 如果不存在相同的记录，则添加进插入列表
				recordsToUpdate = append(recordsToUpdate, item)
			}
		}

		if len(recordsToUpdate) > 0 {
			db.Clauses(clause.OnConflict{
				Columns: []clause.Column{
					{Name: "advertiser_id"},
					{Name: "tdate"},
					{Name: "media_id"},
				},
				DoUpdates: clause.AssignmentColumns([]string{
					"advertiser_name",
					"media_account_id",
					"click",
					"pay_amount_roi",
					"valid_play",
					"stat_pay_amount",
					"cost",
					"convert_cost",
					"cpc",
					"show",
					"ctr",
					"convert",
					"total_play",
					"cpm",
				}),
			}).Create(&recordsToUpdate) // 更新已存在的记录
			updateCount += len(recordsToUpdate) // 增加更新计数
		}

		if len(recordsToInsert) > 0 {
			db.Create(&recordsToInsert)         // 插入新增数据
			insertCount += len(recordsToInsert) // 增加插入计数
		}

	}

	log.Printf("插入数量：%v", insertCount)
	log.Printf("更新数量：%v", updateCount)
}
